package com.ibm.cps.publisher;

import java.math.BigInteger;
import java.util.HashSet;

import org.apache.log4j.Logger;
import org.bson.types.ObjectId;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.newmessage.MetadataInputSchema;
import com.ibm.cps.newmessage.MetadataInputSchema.FieldType;
import com.ibm.interfaces.AbstractMessagePublisher;
import com.ibm.interfaces.IMessageParser;
import com.ibm.interfaces.INoSqlDB;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.mongo.MongoNoSqlDB;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import com.mongodb.MongoClientURI;

public class SeyrenRuleMongoPublisher extends AbstractMessagePublisher implements KryoSerializable{

	private static Logger logger = Logger.getLogger(SeyrenRuleMongoPublisher.class);;
	private INoSqlDB db;
	private String mongo_seyren_url;
	private String mongo_seyren_collection;
	private int dbIndex;
	private String checkId;
	private ArrayNode array;

	public SeyrenRuleMongoPublisher(String metadata) throws CPSException {
		super(metadata);
		ObjectNode node = JSONUtility.fromObjectString(metadata);
		mongo_seyren_url = node.get(MessageFields.MONGO_SEYREN_URL).asText();
		mongo_seyren_collection = node.get(MessageFields.MONGO_SEYREN_COLLECTION).asText();
		MongoClientURI mongoClientURI = new MongoClientURI(mongo_seyren_url);
		if (mongoClientURI.getDatabase() == null) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR,
					"Please provide mongo database in uri " + mongo_seyren_url);
		}
		this.db = new MongoNoSqlDB(mongo_seyren_url, mongoClientURI.getDatabase());
		dbIndex = db.getCollection(mongo_seyren_collection);
		array = JSONUtility.newArrayNode();
		
		checkId = node.get(MessageFields.PUBLISHNAME).asText();
		
	}


	@Override
	public void read(Kryo kryo, Input input) {
		this.mongo_seyren_collection = input.readString();
		this.mongo_seyren_url = input.readString();
		this.checkId = input.readString();
		this.array = kryo.readObject(input, ArrayNode.class);
		MongoClientURI mongoClientURI = new MongoClientURI(mongo_seyren_url);
		this.db = new MongoNoSqlDB(mongo_seyren_url, mongoClientURI.getDatabase());
		try {
			this.dbIndex = db.getCollection(mongo_seyren_collection);
		}catch(CPSException ex){
			throw new IllegalStateException("Failed to connect to db in de-serializing SeyrenRuleMongoPublisher, DB URI=" + mongoClientURI + ", collection=" + mongo_seyren_collection, ex);
		}
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(mongo_seyren_collection);
		output.writeString(mongo_seyren_url);
		output.writeString(checkId);
		kryo.writeObject(output, array);
		try {
			this.db.close();
		} catch (CPSException ex) {
			throw new IllegalStateException("Failed to disconnect db in serializing SeyrenRuleMongoPublisher, DB=" + db, ex);
		}
	}

	@Override
	public void publishMessage(ParsedDataMessage message, IMessageParser parser, boolean isBuffer) throws CPSException {
		ITimeSeriesSchema parentSchema = parser.getSchema();

		String str = new String(parser.restoreFromDataMessage(message));
		ObjectNode node = JSONUtility.fromObjectString(str);
		
		int index = parentSchema.getTagId(MessageFields.RULENAME);
		if (index != -1) {
			String ruleName = (String) message.getValue(index);
			if (ruleName.endsWith("warning")) {
				node.put("alertType", "warning");
			} else if (ruleName.endsWith("error")) {
				node.put("alertType", "error");
			}
		}
		node.put("checkId", checkId.replace("seyren", ""));
		array.add(node);
		if (!isBuffer) {
			ObjectNode result = JSONUtility.newObjectNode();
			String id = new BigInteger(new ObjectId().toByteArray()).toString(16);
			result.put("id", id);
			result.set("message", array);
			HashSet<String> tagset = new HashSet<String>();
			tagset.add("id");
			tagset.add("message");
			db.upsert(dbIndex, result, "id", tagset);
			logger.info("insert to mongo :" + result.toString());
			array.removeAll();
		}
	}

	@Override
	public MetadataInputSchema getInputSchema() {
		MetadataInputSchema schema = new MetadataInputSchema();
		schema.addField(FieldType.String, MessageFields.MONGO_SEYREN_URL, MessageFields.MONGO_SEYREN_COLLECTION,MessageFields.PUBLISHNAME);
		
		return schema;
	}

	@Override
	public void close() {

	}

}
